package mr.diff;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.Iterator;

import mr.CReducer;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import util.DiffComputation;
import util.DiffMessage2;
import util.Pair;
import util.Parsable;
import util.ScanDList;
import util.Utilties;

//reduce
public class DiffStep3<T0 extends Parsable<T0>, T1 extends Parsable<T1>, T2 extends Parsable<T2>>
		extends CReducer<T0, T1, T2, DiffMessage2<T1, T2>> {

	private static final Log LOG = LogFactory.getLog(DiffStep3.class);

	T2 accSum;
	T1 accTims;
	int beginFlg;

	T0 insT0;
	T2 pacc;
	Utilties<T1> tl1;

	Utilties<T2> tl2;

	public DiffStep3(DiffComputation<T0, T1, T2> di) {
		super(di);

		accTims = otimes.id();
		accSum = oplus.id();
		pacc = null;

		insT0 = p.getInstanceOfT0();
		tl1 = new Utilties<T1>();
		tl2 = new Utilties<T2>();

		beginFlg = 0;
	}

	@Override
	public void reduce(LongWritable key, Iterator<Text> values,
			OutputCollector<Text, NullWritable> output, Reporter reporter)
			throws IOException {
		ScanDList<T2> psums = new ScanDList<T2>();
		ScanDList<T1> ptimes = new ScanDList<T1>();

		int countMsg = 0;
		DiffMessage2<T1, T2> receiver = new DiffMessage2<T1, T2>(
				accTims.getClass(), accSum.getClass());

		while (values.hasNext()) {

			Text msg = values.next();

			countMsg++;
			Pair<T1, T2> previousAcc = new Pair<T1, T2>();
			try {
				previousAcc = receiver.parse(msg.toString()).getSecond(); // @@@
			} catch (NoSuchMethodException | SecurityException
					| ClassNotFoundException | InstantiationException
					| IllegalAccessException | IllegalArgumentException
					| InvocationTargetException e) {
				e.printStackTrace();
			}
			// System.out.println("DiffStep3 DiffMessage2: " +
			// previousAcc.getFirst());
			ptimes.put(receiver.getMsgId(), previousAcc.getFirst());

			// System.out.println("DiffStep3 DiffMessage2: " +
			// previousAcc.getSecond());
			psums.put(receiver.getMsgId(), previousAcc.getSecond());

		}

		if (countMsg > 0) {

			accTims = Utilties.fold(ptimes, otimes);
			accSum = Utilties.fold(psums, oplus);

			// System.out.println("step 3, foldl accTims" + accTims);

			// System.out.println("step 3, foldl  accSum" + accSum);

			T2 result = oplus.evaluate(accSum, g.evaluate(accTims));

			System.out.println("Fianl result: " + result);
			output.collect(new Text(result.toString()), NullWritable.get());
			LOG.info("===final result===:" + result.toString());

		}else{
			LOG.warn("Something wrong, no output");
		}
	}

}